/*  File            kvsd.c
 *  --------------------------------------------
 *  Version         2.0
 *  Author          qyorewang@gmail.com
 *  modification    Concurrent model using epoll
 *  --------------------------------------------
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <sys/epoll.h>
#include <assert.h>
#include <arpa/inet.h>
#include <kvs.h>

#define MAXEVENTS       1024
#define BACKLOG         128 
#define MAXEPOLLS       256

#define COMMAND_WAIT    1
#define DATA_WAIT       2
#define WRITE_WAIT      3
#define CLOSE_WAIT      4


#define GET             0X55
#define PUT             0XAA
#define DELETE          0XFF

typedef struct
{
    int type;
    int key_length;
    int value_length;
    int filling;
}PACKET_HEADER;

typedef struct event_handler
{
    int connect_fd;
    int epoll_fd;
    char *buffer;
    int buffsize;
    int buffpos;
    int state;
    int operation;
    char command[16];
}EH, *EHPTR;


static void setnonblock(int fd);
static void init_event_handler(EHPTR ehptr, int cfd, int efd);


int deal_listen(int port);
int deal_accept(int listen_fd);

int deal_parse(EHPTR ehptr);
int deal_read(EHPTR ehptr);
int deal_write(EHPTR ehptr);

int main()
{
    int listen_fd, connect_fd, epoll_fd, log_fd, nfds;
    struct epoll_event ev, events[MAXEVENTS];
    EHPTR ehptr;

    KVS_SET kvs;
    close(2);                                       //close stderr
    freopen("server.log", "a+", stderr);     
    kvs.init_type=KVS_CREATE;                       // or KVS_LOAD if load kvs.
    kvs.size_in_g=1000;
    strcpy(kvs.disk_file, "/tmp/disk");
    strcpy(kvs.log, "/var/log/kvs.log");
    strcpy(kvs.mem_log, "/var/log/kvs_mp.log");
    strcpy(kvs.disk_log, "/var/log/kvs_buddy.log");
    strcpy(kvs.sync_log, "/var/log/kvs_sync.log");
    strcpy(kvs.index_log, "/var/log/kvs_index.log");
    strcpy(kvs.bitmap, "bitmap");
    log_fd=open("KvsServer.log", O_CREAT|O_APPEND);
    kv_init_2(&kvs);

    listen_fd = deal_listen(8080);

    if ((epoll_fd = epoll_create(MAXEPOLLS)) < 0)
    {
        perror("epoll_create error");
        exit(-1);
    }
    
    ev.events = EPOLLIN;
    ev.data.fd = listen_fd;

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev) == -1)
    {
        perror("epoll_ctl add error");
        exit(-1);
    }

    for(;;)
    {
        if ((nfds = epoll_wait(epoll_fd, events, MAXEVENTS, -1)) == -1)
        {
            perror("epoll_wait error");
            exit(-1);
        }
        
        int i;

        for (i = 0; i < nfds; i++)
        {
            fprintf(stderr, "EVENT-data:%d EVENT-events:%d\n", events[i].data.fd, events[i].events);
            fflush(stderr);
            if (events[i].data.fd == listen_fd)
            {
                fprintf(stderr, "EVENT FROM listenfd\n");
                fflush(stderr);
                ehptr = (EHPTR)malloc(sizeof(EH));
                connect_fd = deal_accept(listen_fd);
                init_event_handler(ehptr, connect_fd, epoll_fd);
                ev.data.ptr = (void *)ehptr;
                ev.events = EPOLLIN|EPOLLET;
                assert(epoll_ctl(epoll_fd, EPOLL_CTL_ADD, connect_fd, &ev) != -1);
            }
            else if (events[i].events&EPOLLIN)
            {
                ehptr = (EHPTR)events[i].data.ptr;
                if (ehptr->state==COMMAND_WAIT)
                {
                    fprintf(stderr, "Read command from %d\n", ehptr->connect_fd);
                    fflush(stderr);
                    deal_parse(ehptr);
                }

                else if (ehptr->state==DATA_WAIT)
                {
                    fprintf(stderr, "Read data from %d\n", ehptr->connect_fd);
                    fflush(stderr);
                    deal_read(ehptr);
                }
            }
            else if (events[i].events&EPOLLOUT)
            {
                fprintf(stderr, "Write data to %d\n", ehptr->connect_fd);
                fflush(stderr);
                ehptr=(EHPTR)events[i].data.ptr;
                deal_write(ehptr);
            }
        }
    } 
    return 0;
}

int deal_listen(int port)
{
    int listen_fd;
    struct sockaddr_in servaddr;

    if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
    {
        perror("open socket file error");
        exit(-1);
    }

    int flag = 1;

    if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, (const void *)&flag, sizeof(flag)) == -1)
    {
        perror("setsockopt error");
        exit(-1);
    }

    setnonblock(listen_fd);

    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);

    if (bind(listen_fd, (struct sockaddr *)&servaddr, sizeof(servaddr)) < 0)
    {
        perror("bind error");
        exit(-1);
    }

    if (listen(listen_fd, BACKLOG) < 0)
    {
        perror("listen error");
        exit(-1);
    }

    return listen_fd;
}

int deal_accept(int listen_fd)
{
    int connect_fd;
    struct sockaddr_in client_addr;
    socklen_t addr_len;

    addr_len= sizeof(client_addr);

    if ((connect_fd = accept(listen_fd, (struct sockaddr*)&client_addr, &addr_len)) < 0)
    {
        perror("accept error");
        exit(-1);
    } 

    fprintf(stderr, "Client IP:%s Port:%d sockfd:%d\n", inet_ntoa(client_addr.sin_addr), client_addr.sin_port, connect_fd);
    fflush(stderr);
    setnonblock(connect_fd);
    return connect_fd;
}

int deal_parse(EHPTR ehptr)
{
    int efd = ehptr->epoll_fd;
    int cfd = ehptr->connect_fd;
    PACKET_HEADER *ph = (PACKET_HEADER *)ehptr->command;
    int n;

    n = read(cfd, (void *)ph, sizeof(PACKET_HEADER));
    fprintf(stderr, "Read command %d bytes\n", n);
    fflush(stderr);
    if (n > 0)
        assert(n==16);
    else 
    {
        epoll_ctl(efd, EPOLL_CTL_DEL, cfd, NULL);
        close(cfd);
        free((void *)ehptr);
        return -1;
    }

    if (ph->type==PUT)
    {
        fprintf(stderr, "Command:PUT\n");
        fflush(stderr);
        ehptr->operation=PUT;
        ehptr->buffsize=ph->key_length+ph->value_length;
    }
    else if (ph->type==GET)
    {
        fprintf(stderr, "Command:GET\n");
        fflush(stderr);
        ehptr->operation=GET;
        ehptr->buffsize=ph->key_length;
    }
    else if (ph->type==DELETE)
    {
        fprintf(stderr, "Command:DELETE\n");
        fflush(stderr);
        ehptr->operation=DELETE;
        ehptr->buffsize=ph->key_length;
    }

    ehptr->buffer=(char *)malloc(ehptr->buffsize);

    if (ehptr->buffer == NULL)
        return -1;
    memset(ehptr->buffer, 0, ehptr->buffsize);
    ehptr->state=DATA_WAIT;

    return 0;
}

int deal_read(EHPTR ehptr)
{
    int cfd = ehptr->connect_fd;
    int efd = ehptr->epoll_fd;
    int toread = ehptr->buffsize;
    int n;
    struct epoll_event ev;

    PACKET_HEADER *ph;
    n = read(cfd, ehptr->buffer+ehptr->buffpos, toread);
    fprintf(stderr, "read %d bytes from cfd\n", n);
    fflush(stderr);
    if (n > 0)
    {
        ehptr->buffpos += n;
        if (ehptr->buffpos >= toread)
        {   
            if (ehptr->operation==GET) 
            {
                char *value=(char *)malloc(1024);
                memset(value, 0, 1024);
                char k[1024];
                memset(k, 0, sizeof(k));
                strncpy(k, ehptr->buffer, ehptr->buffsize);

                if (kv_get(k,ehptr->buffsize,value+4,1020) != 0)
                {
                    memset(value, 0, sizeof(value));
                }
                else
                {
                    *(int*)value = strlen(value+4);
                    ;
                }
                free(ehptr->buffer);
                ehptr->buffer=value;
                ehptr->buffsize=*(int *)value + 4;
                ehptr->buffpos=0;
                ehptr->state=WRITE_WAIT;

                ev.data.ptr=(void *)ehptr;

                ev.events=EPOLLOUT|EPOLLET;
                epoll_ctl(efd, EPOLL_CTL_MOD, cfd, &ev);
            }
            else if(ehptr->operation==PUT)
            {
                ph = (PACKET_HEADER *)(ehptr->command);

                int key_length = ph->key_length;
                int value_length= ph->value_length;
                char k[1024];
                char v[1024];
                memset(k, 0, sizeof(k));
                memset(v, 0, sizeof(v));
                strncpy(k, ehptr->buffer, key_length);
                strncpy(v, ehptr->buffer+key_length, value_length); 

                if (kv_put(k,key_length, v, value_length)==0)
                    ;
                else 
                    ;

                free(ehptr->buffer);
                ehptr->buffer=NULL;
                ehptr->buffsize=0;
                ehptr->buffpos=0;
                ehptr->state=COMMAND_WAIT;
            }
            else if (ehptr->operation==DELETE)
            {
                ph = (PACKET_HEADER *)(ehptr->command);
                int key_length = ph->key_length;   
                char k[1024];
                memset(k, 0, sizeof(k));
                strncpy(k, ehptr->buffer, ehptr->buffsize);

                if (kv_delete(k,key_length) == 0)
                    ;
                else 
                    ;

                free(ehptr->buffer);
                ehptr->buffsize=0;
                ehptr->buffpos=0;
                ehptr->state=COMMAND_WAIT;
            }
        }
    }
    else if (n <= 0)
        return -1;

    return 0;
}

int deal_write(EHPTR ehptr)
{
    assert(ehptr!=NULL);
    int cfd = ehptr->connect_fd;
    int efd = ehptr->epoll_fd;
    int towrite = ehptr->buffsize;
    int n;
    struct epoll_event ev;
    n = write(cfd, ehptr->buffer+ehptr->buffpos, ehptr->buffsize);
    fprintf(stderr, "Write %d bytes to %d\n", n, cfd);
    fflush(stderr);
    if (n > 0)
    {
        ehptr->buffpos+=n;
        if (ehptr->buffpos>=towrite)
        {
            free(ehptr->buffer);
            ehptr->buffer=NULL;
            ehptr->buffpos=0;
            ehptr->buffsize=0;
            ehptr->state=COMMAND_WAIT;
            ev.data.ptr=(void *)ehptr;
            ev.events=EPOLLIN|EPOLLET;
            assert(epoll_ctl(efd, EPOLL_CTL_MOD, cfd, &ev)!=-1);
        }
    }
    return 0;
}

static void setnonblock(int fd)
{
    int opt = fcntl(fd, F_GETFL, 0);

    if (fcntl(fd, F_SETFL, opt|O_NONBLOCK) == -1)
    {
        perror("fcntl set failed");
        exit(-1);
    }
}

static void init_event_handler(EHPTR ehptr, int cfd, int efd)
{
    assert(ehptr != NULL);

    ehptr->connect_fd = cfd;
    ehptr->epoll_fd = efd;
    memset(ehptr->command, 0, sizeof(ehptr->command));
    ehptr->buffer=NULL;
    ehptr->buffpos=0;
    ehptr->buffsize=0;
    ehptr->state=COMMAND_WAIT;
}
void event_log()
{
}
